package org.hw.util;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.json.JSONException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.json.JSONArray;
import org.json.JSONObject;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class MqttUtil {
    private static MqttClient client;
    private static final String BROKER = "tcp://47.108.153.126:1883";
    private static final String CLIENT_ID = "javaClient";
    private static final String USERNAME = "admin";
    private static final String PASSWORD = "xf123456";
    private static final int CONNECTION_TIMEOUT = 30;
    private static final int KEEP_ALIVE_INTERVAL = 60;
    // 存储主题和对应回调的映射
    private static final Map<String, MqttCallback> topicCallbacks = new HashMap<>();
    // 存储已订阅的主题
    private static final Set<String> subscribedTopics = new HashSet<>();
    private static final ConcurrentLinkedQueue<String> messageQueue = new ConcurrentLinkedQueue<>();

    private static final Logger logger = LoggerFactory.getLogger(MqttUtil.class);
    /**
     * 初始化 MQTT 客户端连接
     */
    public static void init() {
        try {
            MemoryPersistence persistence = new MemoryPersistence();
            client = new MqttClient(BROKER, CLIENT_ID, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setConnectionTimeout(CONNECTION_TIMEOUT);
            connOpts.setKeepAliveInterval(KEEP_ALIVE_INTERVAL);
            connOpts.setCleanSession(true);
            if (USERNAME != null && !USERNAME.isEmpty()) {
                connOpts.setUserName(USERNAME);
                connOpts.setPassword(PASSWORD.toCharArray());
            }
            client.connect(connOpts);
            System.out.println("MQTT 客户端连接成功");
        } catch (MqttException e) {
            e.printStackTrace();
            System.err.println("MQTT 客户端连接失败: " + e.getMessage());
        }
    }

    /**
     * 发布消息到指定主题
     * @param topic 主题
     * @param message 消息内容
     * @param qos 服务质量（0、1、2）
     */
    public static void publish(String topic, String message, int qos) {
        if (client == null ||!client.isConnected()) {
            init();
        }
        try {
            MqttMessage mqttMessage = new MqttMessage(message.getBytes());
            mqttMessage.setQos(qos);
            client.publish(topic, mqttMessage);
            System.out.println("已发布消息到主题: " + topic + "，内容: " + message);
        } catch (MqttException e) {
            e.printStackTrace();
            System.err.println("发布消息失败: " + e.getMessage());
        }
    }

    /**
     * 订阅多个主题
     * @param topics 要订阅的主题数组
     * @param qos 服务质量（0、1、2）
     * @param callback 消息回调处理类，处理收到的消息
     */
    public static void subscribeMultiple(String[] topics, int qos, MqttCallback callback) {
        if (client == null ||!client.isConnected()) {
            init();
        }
        try {
            // 设置统一的回调处理器
            client.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    System.err.println("连接丢失，原因: " + cause.getMessage());
                    // 自动重连逻辑
                    new Thread(() -> {
                        int retryCount = 0;
                        while (!client.isConnected() && retryCount < 10) {
                            try {
                                logger.info("尝试重新连接MQTT服务器，第{}次", retryCount + 1);
                                client.reconnect();
                                logger.info("MQTT重新连接成功");
                                // 重新订阅之前的主题
                                resubscribeTopics();
                                break;
                            } catch (MqttException e) {
                                retryCount++;
                                logger.error("MQTT重连失败，原因: {}", e.getMessage());
                                try {
                                    Thread.sleep(1000 * (retryCount + 1)); // 指数退避重试
                                } catch (InterruptedException ie) {
                                    Thread.currentThread().interrupt();
                                    return;
                                }
                            }
                        }
                    }).start();
                    // 通知所有回调连接丢失
                    for (MqttCallback cb : topicCallbacks.values()) {
                        cb.connectionLost(cause);
                    }
                }
                // 添加重新订阅主题的方法
                private void resubscribeTopics() throws MqttException {
                    if (subscribedTopics.isEmpty()) return;
                    String[] topics = subscribedTopics.toArray(new String[0]);
                    int[] qosLevels = new int[topics.length];
                    Arrays.fill(qosLevels, 1); // 假设使用QoS 1
                    client.subscribe(topics, qosLevels);
                    logger.info("已重新订阅主题: {}", String.join(", ", topics));
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    // 将消息添加到队列
                    String payload = new String(message.getPayload());
                    messageQueue.add(payload);
                    // 查找对应主题的回调并处理消息
                    MqttCallback cb = topicCallbacks.get(topic);
                    if (cb != null) {
                        cb.messageArrived(topic, message);
                    }
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    // 通知所有回调消息交付完成
                    for (MqttCallback cb : topicCallbacks.values()) {
                        cb.deliveryComplete(token);
                    }
                }
            });

            int[] qosLevels = new int[topics.length];
            for (int i = 0; i < topics.length; i++) {
                qosLevels[i] = qos;
                topicCallbacks.put(topics[i], callback);
                // 将主题添加到已订阅主题集合中
                subscribedTopics.add(topics[i]);
            }
            client.subscribe(topics, qosLevels);
            System.out.println("已订阅主题: " + String.join(", ", topics));
        } catch (MqttException e) {
            e.printStackTrace();
            System.err.println("订阅主题失败: " + e.getMessage());
        }
    }

    /**
     * 获取所有已订阅的主题
     * @return 已订阅主题的集合
     */
    public static Set<String> getSubscribedTopics() {
        return new HashSet<>(subscribedTopics);
    }

    /**
     * 关闭 MQTT 客户端连接
     */
    public static void close() {
        if (client != null && client.isConnected()) {
            try {
                client.disconnect();
                client.close();
                System.out.println("MQTT 客户端连接已关闭");
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }

    private static final String TOPIC = "test/topic";
    private static final int QOS = 0;

    /**
     * 从消息队列中获取所有消息，并清空队列
     * @return 包含队列中所有消息的列表，如果队列为空则返回空列表
     */
    public static List<String> getMessagesFromQueue() {
        List<String> messages = new ArrayList<>();
        String message = "";
        try {
            // 遍历队列并移除所有元素
            while ((message = messageQueue.poll()) != null) {
                messages.add(message);
            }
            logger.info("从消息队列中获取到 {} 条消息", messages.size());
        } catch (Exception e) {
            logger.error("获取消息队列中的消息时发生错误", e);
            messages.add(message);
        }
        return messages;
    }

    /**
     * 从 EMQX 代理获取所有主题
     * @return 所有主题的列表
     * @throws IOException 网络请求异常
     */
    public static List<String> getAllTopicsFromEmqx() throws IOException {
        HttpClient httpClient = HttpClients.createDefault();
        HttpGet httpGet = new HttpGet("http://123.56.3.98:18083/api/v5/stats/topics");
        httpGet.setHeader("Authorization", "Basic " + java.util.Base64.getEncoder().encodeToString((USERNAME + ":" + PASSWORD).getBytes()));

        HttpResponse response = httpClient.execute(httpGet);
        String responseBody = EntityUtils.toString(response.getEntity());

        JSONObject jsonResponse = null;
        List<String> topics = new ArrayList<>();
        try {
            jsonResponse = new JSONObject(responseBody);
            JSONArray topicsArray = null;
            topicsArray = jsonResponse.getJSONArray("data");
            for (int i = 0; i < topicsArray.length(); i++) {
                topics.add(topicsArray.getString(i));
            }
        } catch (JSONException e) {
            throw new RuntimeException(e);
        }
        return topics;
    }

}